// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: daniel

#include "coin2/base/request/http2/session_pool.h"

namespace coin2::base::request::http2 {

// default ssl context is a global instance (boost::asio::ssl::context::sslv23_client)
// ssl context may be changed by nghttp2
// so here use local instance (boost::asio::ssl::context::sslv23)
SessionPool::SessionPool(boost::asio::io_context* io_ctx)
    : io_ctx_{io_ctx},
      ssl_ctx_{boost::asio::ssl::context::sslv23},
      timer_{*io_ctx_},
      max_sessions_per_host_{0} {}

SessionPool::SessionPool(boost::asio::io_context* io_ctx, int max_sessions_per_host)
    : io_ctx_{io_ctx},
      ssl_ctx_{boost::asio::ssl::context::sslv23},
      timer_{*io_ctx_},
      max_sessions_per_host_{max_sessions_per_host} {}

SessionPool::SessionPool(
    boost::asio::io_context* io_ctx,
    boost::asio::ssl::context* ssl_ctx,
    int max_sessions_per_host)
    : io_ctx_{io_ctx},
      ssl_ctx_{boost::asio::ssl::context::sslv23},
      timer_{*io_ctx_},
      max_sessions_per_host_{max_sessions_per_host} {}

void SessionPool::AsyncRequest(
    const std::string& host,
    int port,
    const std::string& bind_ip,
    RequestType req,
    CallbackType callback,
    RequestPriorityWeight priority_weight) {
  auto session = GetSession(host, port, bind_ip);
  if (!session) {
    LOG(ERROR) << "[http2] failed to send request due to no available session with "
               << "host=" << host << ", port=" << port << ", bind_ip=" << bind_ip;
    auto ts = GetCurrentTimestamp();
    using SessionResponse = coin2::base::request::SessionResponse;
    SessionResponse response;
    response.error_code =
        boost::system::errc::make_error_code(boost::system::errc::too_many_files_open);
    response.raw_request = &req;
    response.request_ts = ts;
    response.response_ts = ts;
    callback(response.error_code, &response);
    return;
  }
  (*session)->AsyncRequest(req, callback, priority_weight);
}

void SessionPool::StartTimer() {
  timer_enabled_ = true;
  timer_.expires_after(std::chrono::seconds(timer_interval_sec_));
  timer_.async_wait(std::bind(&SessionPool::OnTimer, shared_from_this(), std::placeholders::_1));
}

auto SessionPool::GetSession(const std::string& host, int port, const std::string& bind_ip)
    -> std::optional<Http2Session*> {
  if (host.empty() || port == 0) {
    return {};
  }
  Key key(host, port, bind_ip);
  if (auto it = pools_.find(key); it != pools_.end()) {
    return it->second.get();
  }
  // TODO(daniel) - check by host
  if (max_sessions_per_host_ > 0 && sessions_per_host_.count(host) > 0 &&
      sessions_per_host_.at(host) >= max_sessions_per_host_) {
    LOG(ERROR) << "max session per host limit hit";
    return {};
  }
  auto session = std::make_unique<Http2Session>(io_ctx_, &ssl_ctx_, host, port, bind_ip);
  if (!session) {
    LOG(ERROR) << "failed to create http2 session";
    return {};
  }
  // session per host
  if (sessions_per_host_.count(host) == 0) {
    sessions_per_host_[host] = 1;
  } else {
    ++sessions_per_host_[host];
  }
  auto ret = session.get();
  pools_.emplace(key, std::move(session));
  return ret;
}

void SessionPool::OnTimer(const boost::system::error_code& ec) {
  if (ec || !timer_enabled_) {
    return;
  }
  if (timeout_ns_ > 0) {
    for (auto& pair : pools_) {
      pair.second->ProcessTimeout(timeout_ns_);
    }
  }
  timer_.expires_after(std::chrono::seconds(timer_interval_sec_));
  timer_.async_wait(std::bind(&SessionPool::OnTimer, shared_from_this(), std::placeholders::_1));
}

}  // namespace coin2::base::request::http2
